/**
 * @file test_concurrency.cpp
 * @brief 消息并发性能测试
 *
 * 测试目标：
 * - 多Producer并发发送：无消息丢失
 * - 多Consumer并发消费：负载均衡
 * - 高并发场景：系统稳定性
 */

#include <gtest/gtest.h>
#include <spdlog/spdlog.h>
#include <chrono>
#include <vector>
#include <set>
#include <atomic>
#include <thread>
#include <iostream>
#include <iomanip>
#include "MessageProducer.h"
#include "MessageConsumer.h"
#include "MessageDispatcher.h"
#include "../test_helpers.h"

using namespace std::chrono;
using namespace MessageFramework;
using namespace MessageFramework::Test;

/**
 * @brief 并发测试Handler
 */
class ConcurrencyHandler : public IMessageHandler {
public:
    int handleMessage(const Message& message) override {
        std::lock_guard<std::mutex> lock(mutex_);

        // 记录消息ID
        std::string msgId = message.header.messageId;
        receivedIds_.insert(msgId);

        // 记录消费者ID
        consumerStats_[message.header.consumerId]++;

        count_++;
        return 0;  // Success
    }

    std::string getMessageType() const override {
        return "ConcurrencyTest";
    }

    std::string getName() const override {
        return "ConcurrencyHandler";
    }

    int getCount() const {
        std::lock_guard<std::mutex> lock(mutex_);
        return count_;
    }

    std::set<std::string> getReceivedIds() const {
        std::lock_guard<std::mutex> lock(mutex_);
        return receivedIds_;
    }

    std::map<std::string, int> getConsumerStats() const {
        std::lock_guard<std::mutex> lock(mutex_);
        return consumerStats_;
    }

    void reset() {
        std::lock_guard<std::mutex> lock(mutex_);
        receivedIds_.clear();
        consumerStats_.clear();
        count_ = 0;
    }

private:
    mutable std::mutex mutex_;
    std::set<std::string> receivedIds_;
    std::map<std::string, int> consumerStats_;
    std::atomic<int> count_{0};
};

/**
 * @brief 并发性能测试
 */
class ConcurrencyPerformanceTest : public ::testing::Test {
protected:
    void SetUp() override {
        TestHelpers::cleanupRedis("test_concurrency_*");
        pool_ = TestHelpers::getTestRedisPool();
        streamName_ = "test_concurrency_stream_" + TestHelpers::randomString(8);
        handler_ = std::make_shared<ConcurrencyHandler>();
    }

    void TearDown() override {
        // 停止所有Consumer
        for (auto& consumer : consumers_) {
            if (consumer) consumer->stop();
        }
        consumers_.clear();

        TestHelpers::cleanupRedis("test_concurrency_*");
    }

    /**
     * @brief 创建Producer
     */
    std::unique_ptr<MessageProducer> createProducer(const std::string& publisherId) {
        MessageProducer::Config config;
        config.publisherId = publisherId;
        config.autoSign = false;
        config.autoEncrypt = false;
        return std::make_unique<MessageProducer>(pool_, config);
    }

    /**
     * @brief 创建Consumer
     */
    std::unique_ptr<MessageConsumer> createConsumer(const std::string& consumerId) {
        MessageConsumer::Config config;
        config.streamName = streamName_;
        config.consumerGroup = "ConcurrencyGroup";
        config.consumerId = consumerId;
        config.blockMs = 10;
        config.batchSize = 10;

        auto consumer = std::make_unique<MessageConsumer>(pool_, config);

        // 配置Dispatcher
        MessageDispatcher::Config dispatcherConfig;
        dispatcherConfig.verifySignature = false;
        dispatcherConfig.autoDecrypt = false;
        auto dispatcher = std::make_shared<MessageDispatcher>(dispatcherConfig);
        dispatcher->registerHandler("ConcurrencyTest", handler_);

        // 设置回调
        consumer->setMessageCallback([dispatcher](const Message& msg) {
            auto result = dispatcher->dispatch(const_cast<Message&>(msg));
            return result.code;
        });

        consumers_.push_back(std::move(consumer));
        return std::move(consumers_.back());
    }

    RedisPool* pool_;
    std::string streamName_;
    std::shared_ptr<ConcurrencyHandler> handler_;
    std::vector<std::unique_ptr<MessageConsumer>> consumers_;
};

/**
 * @brief 测试多Producer并发发送
 */
TEST_F(ConcurrencyPerformanceTest, MultiProducerConcurrent) {
    std::cout << "\n========== 多Producer并发测试 (3个Producer，各发送1000条) ==========\n";

    const int producerCount = 3;
    const int messagesPerProducer = 1000;
    const int totalMessages = producerCount * messagesPerProducer;

    // 启动1个Consumer
    auto consumer = createConsumer("Consumer1");
    consumer->start();
    std::this_thread::sleep_for(milliseconds(100));

    // 创建多个Producer并发送消息
    std::vector<std::thread> threads;
    std::atomic<int> totalSent{0};

    auto startTime = high_resolution_clock::now();

    for (int i = 0; i < producerCount; i++) {
        threads.emplace_back([this, i, messagesPerProducer, &totalSent]() {
            auto producer = createProducer("Producer" + std::to_string(i));

            for (int j = 0; j < messagesPerProducer; j++) {
                std::string body = "Producer" + std::to_string(i) + "_Msg" + std::to_string(j);
                producer->send(streamName_, "ConcurrencyTest", body);
                totalSent++;
            }
        });
    }

    // 等待所有Producer完成
    for (auto& thread : threads) {
        thread.join();
    }

    auto endTime = high_resolution_clock::now();
    auto duration = duration_cast<milliseconds>(endTime - startTime).count();

    std::cout << "发送完成: " << totalSent << " 条消息\n";
    std::cout << "发送耗时: " << duration << " ms\n";
    std::cout << "发送速率: " << (totalSent * 1000 / duration) << " msg/s\n";

    // 等待所有消息被消费
    bool allReceived = TestHelpers::waitFor([this, totalMessages]() {
        return handler_->getCount() >= totalMessages;
    }, 30000);

    ASSERT_TRUE(allReceived) << "未能在30秒内接收所有消息";

    // 验证消息完整性
    auto receivedIds = handler_->getReceivedIds();
    EXPECT_EQ(receivedIds.size(), totalMessages) << "应该接收到所有不重复的消息";
    EXPECT_EQ(handler_->getCount(), totalMessages) << "消息计数应该正确";
}

/**
 * @brief 测试多Consumer并发消费
 */
TEST_F(ConcurrencyPerformanceTest, MultiConsumerConcurrent) {
    std::cout << "\n========== 多Consumer并发测试 (1个Producer，3个Consumer) ==========\n";

    const int consumerCount = 3;
    const int totalMessages = 3000;

    // 启动多个Consumer
    for (int i = 0; i < consumerCount; i++) {
        auto consumer = createConsumer("Consumer" + std::to_string(i));
        consumer->start();
    }
    std::this_thread::sleep_for(milliseconds(100));

    // 创建Producer并发送消息
    auto producer = createProducer("Producer1");

    auto startTime = high_resolution_clock::now();
    for (int i = 0; i < totalMessages; i++) {
        std::string body = "Message_" + std::to_string(i);
        producer->send(streamName_, "ConcurrencyTest", body);
    }
    auto sendEndTime = high_resolution_clock::now();

    auto sendDuration = duration_cast<milliseconds>(sendEndTime - startTime).count();
    std::cout << "发送耗时: " << sendDuration << " ms\n";
    std::cout << "发送速率: " << (totalMessages * 1000 / sendDuration) << " msg/s\n";

    // 等待所有消息被消费
    bool allReceived = TestHelpers::waitFor([this, totalMessages]() {
        return handler_->getCount() >= totalMessages;
    }, 30000);

    auto receiveEndTime = high_resolution_clock::now();
    auto receiveDuration = duration_cast<milliseconds>(receiveEndTime - sendEndTime).count();

    ASSERT_TRUE(allReceived) << "未能在30秒内接收所有消息";

    std::cout << "接收耗时: " << receiveDuration << " ms\n";
    std::cout << "接收速率: " << (totalMessages * 1000 / receiveDuration) << " msg/s\n";

    // 验证消息完整性
    auto receivedIds = handler_->getReceivedIds();
    EXPECT_EQ(receivedIds.size(), totalMessages) << "应该接收到所有不重复的消息";

    // 验证负载均衡
    auto stats = handler_->getConsumerStats();
    std::cout << "\n负载分布:\n";
    for (const auto& [consumerId, count] : stats) {
        double percentage = (count * 100.0) / totalMessages;
        std::cout << "  " << consumerId << ": " << count << " 条 ("
                  << std::fixed << std::setprecision(1) << percentage << "%)\n";
    }

    // 每个Consumer应该处理一部分消息
    for (const auto& [consumerId, count] : stats) {
        EXPECT_GT(count, 0) << consumerId << " 应该处理至少一条消息";
    }
}

/**
 * @brief 测试多Producer多Consumer并发
 */
TEST_F(ConcurrencyPerformanceTest, MultiProducerMultiConsumer) {
    std::cout << "\n========== 多Producer多Consumer并发测试 ==========\n";
    std::cout << "配置: 5个Producer，5个Consumer，每个Producer发送500条\n\n";

    const int producerCount = 5;
    const int consumerCount = 5;
    const int messagesPerProducer = 500;
    const int totalMessages = producerCount * messagesPerProducer;

    // 启动多个Consumer
    for (int i = 0; i < consumerCount; i++) {
        auto consumer = createConsumer("Consumer" + std::to_string(i));
        consumer->start();
    }
    std::this_thread::sleep_for(milliseconds(200));

    // 创建多个Producer线程并发送消息
    std::vector<std::thread> threads;
    std::atomic<int> totalSent{0};

    auto startTime = high_resolution_clock::now();

    for (int i = 0; i < producerCount; i++) {
        threads.emplace_back([this, i, messagesPerProducer, &totalSent]() {
            auto producer = createProducer("Producer" + std::to_string(i));

            for (int j = 0; j < messagesPerProducer; j++) {
                std::string body = "P" + std::to_string(i) + "_M" + std::to_string(j);
                producer->send(streamName_, "ConcurrencyTest", body);
                totalSent++;
            }
        });
    }

    // 等待所有Producer完成
    for (auto& thread : threads) {
        thread.join();
    }

    auto sendEndTime = high_resolution_clock::now();
    auto sendDuration = duration_cast<milliseconds>(sendEndTime - startTime).count();

    std::cout << "发送完成: " << totalSent << " 条消息\n";
    std::cout << "发送耗时: " << sendDuration << " ms\n";
    std::cout << "发送速率: " << (totalSent * 1000 / sendDuration) << " msg/s\n\n";

    // 等待所有消息被消费
    bool allReceived = TestHelpers::waitFor([this, totalMessages]() {
        return handler_->getCount() >= totalMessages;
    }, 60000);

    auto receiveEndTime = high_resolution_clock::now();
    auto receiveDuration = duration_cast<milliseconds>(receiveEndTime - sendEndTime).count();

    ASSERT_TRUE(allReceived) << "未能在60秒内接收所有消息";

    std::cout << "接收完成: " << handler_->getCount() << " 条消息\n";
    std::cout << "接收耗时: " << receiveDuration << " ms\n";
    std::cout << "接收速率: " << (handler_->getCount() * 1000 / receiveDuration) << " msg/s\n\n";

    // 验证消息完整性
    auto receivedIds = handler_->getReceivedIds();
    EXPECT_EQ(receivedIds.size(), totalMessages) << "应该接收到所有不重复的消息";
    EXPECT_EQ(handler_->getCount(), totalMessages) << "消息计数应该正确";

    // 验证负载均衡
    auto stats = handler_->getConsumerStats();
    std::cout << "负载分布:\n";
    for (const auto& [consumerId, count] : stats) {
        double percentage = (count * 100.0) / totalMessages;
        std::cout << "  " << consumerId << ": " << count << " 条 ("
                  << std::fixed << std::setprecision(1) << percentage << "%)\n";
    }

    // 每个Consumer应该处理一部分消息
    for (const auto& [consumerId, count] : stats) {
        EXPECT_GT(count, 0) << consumerId << " 应该处理至少一条消息";
    }
}

/**
 * @brief 测试高并发压力
 */
TEST_F(ConcurrencyPerformanceTest, HighConcurrencyStress) {
    std::cout << "\n========== 高并发压力测试 ==========\n";
    std::cout << "配置: 10个Producer，10个Consumer，每个Producer发送1000条\n\n";

    const int producerCount = 10;
    const int consumerCount = 10;
    const int messagesPerProducer = 1000;
    const int totalMessages = producerCount * messagesPerProducer;

    // 启动多个Consumer
    for (int i = 0; i < consumerCount; i++) {
        auto consumer = createConsumer("Consumer" + std::to_string(i));
        consumer->start();
    }
    std::this_thread::sleep_for(milliseconds(300));

    // 创建多个Producer线程并发送消息
    std::vector<std::thread> threads;
    std::atomic<int> totalSent{0};
    std::atomic<int> sendErrors{0};

    auto startTime = high_resolution_clock::now();

    for (int i = 0; i < producerCount; i++) {
        threads.emplace_back([this, i, messagesPerProducer, &totalSent, &sendErrors]() {
            try {
                auto producer = createProducer("Producer" + std::to_string(i));

                for (int j = 0; j < messagesPerProducer; j++) {
                    std::string body = "P" + std::to_string(i) + "_M" + std::to_string(j);
                    producer->send(streamName_, "ConcurrencyTest", body);
                    totalSent++;
                }
            } catch (const std::exception& e) {
                sendErrors++;
                std::cerr << "Producer " << i << " error: " << e.what() << "\n";
            }
        });
    }

    // 等待所有Producer完成
    for (auto& thread : threads) {
        thread.join();
    }

    auto sendEndTime = high_resolution_clock::now();
    auto sendDuration = duration_cast<milliseconds>(sendEndTime - startTime).count();

    std::cout << "发送完成: " << totalSent << " 条消息\n";
    std::cout << "发送错误: " << sendErrors << " 次\n";
    std::cout << "发送耗时: " << sendDuration << " ms\n";
    std::cout << "发送速率: " << (totalSent * 1000 / sendDuration) << " msg/s\n\n";

    // 等待所有消息被消费（延长超时时间）
    bool allReceived = TestHelpers::waitFor([this, totalMessages]() {
        int received = handler_->getCount();
        if (received % 1000 == 0) {
            std::cout << "已接收: " << received << " / " << totalMessages << "\r" << std::flush;
        }
        return received >= totalMessages;
    }, 120000);

    auto receiveEndTime = high_resolution_clock::now();
    auto receiveDuration = duration_cast<milliseconds>(receiveEndTime - sendEndTime).count();

    std::cout << "\n";
    ASSERT_TRUE(allReceived) << "未能在120秒内接收所有消息";

    std::cout << "接收完成: " << handler_->getCount() << " 条消息\n";
    std::cout << "接收耗时: " << receiveDuration << " ms\n";
    std::cout << "接收速率: " << (handler_->getCount() * 1000 / receiveDuration) << " msg/s\n\n";

    // 验证消息完整性
    auto receivedIds = handler_->getReceivedIds();
    EXPECT_EQ(receivedIds.size(), totalMessages) << "应该接收到所有不重复的消息";
    EXPECT_EQ(handler_->getCount(), totalMessages) << "消息计数应该正确";
    EXPECT_EQ(sendErrors.load(), 0) << "不应该有发送错误";

    // 验证负载分布
    auto stats = handler_->getConsumerStats();
    std::cout << "负载分布:\n";

    int minLoad = totalMessages;
    int maxLoad = 0;

    for (const auto& [consumerId, count] : stats) {
        double percentage = (count * 100.0) / totalMessages;
        std::cout << "  " << consumerId << ": " << count << " 条 ("
                  << std::fixed << std::setprecision(1) << percentage << "%)\n";

        minLoad = std::min(minLoad, count);
        maxLoad = std::max(maxLoad, count);
    }

    // 负载均衡度检查（最大负载不应超过最小负载的3倍）
    std::cout << "\n负载均衡度: 最小=" << minLoad << ", 最大=" << maxLoad
              << ", 比值=" << (double)maxLoad / minLoad << "\n";

    EXPECT_LT((double)maxLoad / minLoad, 3.0) << "负载应该相对均衡";
}

int main(int argc, char** argv) {
    ::testing::InitGoogleTest(&argc, argv);
    return RUN_ALL_TESTS();
}
